package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.gmall.realtime.app.func.DimSinkFunction;
import com.atguigu.gmall.realtime.app.func.MyDebeziumDeserializationSchema;
import com.atguigu.gmall.realtime.app.func.TableProcessFunction;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * Author: Felix
 * Date: 2022/4/2
 * Desc: 业务数据动态分流
 *
 * 需要启动的进程
 *      zk、kafka、maxwell、HDFS、hbase、BaseDBApp
 *
 *代码编写过程
 *  ------------------业务流读取---------------------
 * 	创建Flink应用程序BaseDBApp
 * 	基本环境
 * 		指定流处理环境
 * 		设置并行度
 * 	检查点相关设置
 * 	从Kafka主题ods_base_db_m中读取业务数据
 * 	对读取的数据进行类型转换   jsonStr->jsonObj
 * 	简单的ETL
 * 	------------------配置表读取---------------------
 * 	使用FlinkCDC读取配置表中数据---配置流
 * 	对读取的配置流进行广播---广播流
 * 	------------------分流业务实现---------------------
 * 	将业务流和广播流进行关联---connect
 * 	对关联之后的数据进行处理（分流）---process---BroadcastProcessFunction
 * 	    封装了一个类 TableProcessFunction
 * 	        processElement---处理业务流数据
 * 	            4.字段的过滤
 * 	            2.从广播状态中获取配置信息，判断处理业务数据是事实还是维度 实现简单的分流
 * 	        processBrocastElement---处理广播流数据
 * 	            1.读取广播流中的配置信息，封装为TableProcess对象，放到广播状态中
 * 	            3.如果是读取到的配置信息是维度配置，提前创建维度的表
 * 	--------------------分别对主流和维度侧输出流进行处理-------------------
 * 	将维度侧输出流数据写到phoenix表中
 * 	    封装了一个类DimSinkFunction
 * 	        invoke{
 * 	            拼接upsert语句
 * 	            使用jdbc方式执行upsert
 * 	        }
 * 	将主流事实数据写到kafka的不同的主题中
 * 	    主流.addSink(
 * 	        new FlinkKafkaProducer(default_topic,序列化过程,props,精准一次性保证)===>MyKafkaUtil.getKafkaSink
 * 	    )
 *
 *
 * 程序执行流程
 *      BaseDBApp应用启动的时候，首先通过FlinkCDC读取到配置表中的配置信息
 *      读到配置信息之后，会将配置信息放到广播状态中保存起来
 *      当我们修改了业务数据库的表的时候
 *      binlog会记录修改的内容
 *      maxwell会从binlog中读取变化的内容，并将其封装为json字符串，发送到kafka的ods_base_db_m主题
 *      BaseDBApp会从ods_base_db_m主题中读取到数据，并对其进行处理
 *      在处理的时候，会从当前处理的数据中获取表名和操作类型，将表名和操作类型拼接为key，到广播状态中获取对应的配置信息
 *      根据配置信息中的sinkType属性，判断是事实还是维度，从而进行分流操作
 *          事实--主流
 *          维度--侧输出流中
 *      分别对不同流的数据进行处理
 *          侧输出流维度数据---写到phoenix不同的表中
 *          主流事实数据---写到kafka不同的主题中
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        /*
        //TODO 2.检查点相关的设置
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        env.setStateBackend(new FsStateBackend("hdfs://xxxxx"));
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */
        //TODO 3.从kafka中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "ods_base_db_m";
        String groupId = "base_db_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaSource);
        //TODO 4.对读取的数据进行类型转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        //TODO 5.简单的ETL
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.filter(
            new FilterFunction<JSONObject>() {
                @Override
                public boolean filter(JSONObject jsonObj) throws Exception {
                    boolean flag = jsonObj.getString("table")!=null &&
                        jsonObj.getString("table").length() > 0 &&
                        jsonObj.getJSONObject("data") != null &&
                        jsonObj.getString("data").length() > 3;
                    return flag;
                }
            }
        );
        //filterDS.print(">>>>>>");

        //TODO 6.使用FlinkCDC读取MySQL中的配置表  得到一条流
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
            .hostname("hadoop202")
            .port(3306)
            .databaseList("gmall0925_realtime")
            .tableList("gmall0925_realtime.table_process")
            .username("root")
            .password("123456")
            .startupOptions(StartupOptions.initial())
            .deserializer(new MyDebeziumDeserializationSchema())
            .build();
        DataStreamSource<String> mySQLDS = env.addSource(sourceFunction);

        //TODO 7.对读取到的配置信息进行广播  得到广播流
        MapStateDescriptor<String, TableProcess> mapStateDescriptor =
            new MapStateDescriptor<String, TableProcess>("mapStateDescriptor",String.class,TableProcess.class);
        BroadcastStream<String> broadcastDS = mySQLDS.broadcast(mapStateDescriptor);

        //TODO 8.将业务流和广播流（配置）关联在一起  使用connect
        BroadcastConnectedStream<JSONObject, String> connectDS = filterDS.connect(broadcastDS);

        //TODO 9.对关联之后的数据进行处理   分流： 维度数据---维度侧输出流中      事实数据---主流
        OutputTag<JSONObject> dimTag = new OutputTag<JSONObject>("dimTag") {};

        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(
            new TableProcessFunction(dimTag,mapStateDescriptor)
        );

        DataStream<JSONObject> dimDS = realDS.getSideOutput(dimTag);
        realDS.print(">>>");
        dimDS.print("###");

        //TODO 10.将维度侧输出流的数据写到phoenix表中
        dimDS.addSink(new DimSinkFunction());

        //TODO 11.将主流事实数据写到kafka不同主题中
        realDS.addSink(
            MyKafkaUtil.<JSONObject>getKafkaSinkSchema(
                new KafkaSerializationSchema<JSONObject>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, @Nullable Long timestamp) {
                        String topic = jsonObj.getString("sink_table");
                        return new ProducerRecord<byte[], byte[]>(topic,jsonObj.getJSONObject("data").toJSONString().getBytes());
                    }
                }
            )
        );
        env.execute();
    }
}
